package com.ionehe.group.common.notify.rocketmq.producer;

import com.beust.jcommander.Strings;
import com.ionehe.group.common.notify.autoconfigure.AsyncVendors;
import com.ionehe.group.common.notify.configure.NotifyConfiguration;
import com.ionehe.group.common.notify.message.exception.AsyncException;
import com.ionehe.group.common.notify.message.producer.OrderProducer;
import com.ionehe.group.common.notify.rocketmq.flags.RocketKeyFlags;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Copyright (c) 2021 ionehe.com
 * Date: 2021/4/21
 * Time: 1:22 下午
 *
 * @author 2021年 <a href="mailto:a@ionehe.com">秀</a>
 * 【rocketmq生产者(顺序性消息)】
 */
public class OrderProducerImpl implements OrderProducer {
    private final DefaultMQProducer defaultMQProducer;
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final AtomicBoolean closed = new AtomicBoolean(false);


    public OrderProducerImpl(NotifyConfiguration conf) {
        Map<String, Object> vendor = (Map<String, Object>) conf.getVendors().get(AsyncVendors.rocketmq);
        this.defaultMQProducer = new DefaultMQProducer((String) RocketKeyFlags.FLAG_PRODUCER_GROUP.get(vendor));
        this.defaultMQProducer.setSendMsgTimeout(((Integer) RocketKeyFlags.FLAG_SEND_MSG_TIMEOUT.get(vendor)).intValue());
        this.defaultMQProducer.setInstanceName((String) RocketKeyFlags.FLAG_INSTANCE_NAME.get(vendor));
        String nameServer = (String) RocketKeyFlags.FLAG_NAMESRVADDR.get(vendor);
        if (!Strings.isStringEmpty(nameServer)) {
            this.defaultMQProducer.setNamesrvAddr(nameServer);
        }
    }


    @Override
    public void start() {
        try {
            if (this.started.compareAndSet(false, true)) {
                this.defaultMQProducer.start();
            }
        } catch (Exception e) {
            throw new AsyncException(e.getMessage(), e);
        }
    }


    @Override
    public void shutdown() {
        if (this.closed.compareAndSet(false, true)) {
            this.defaultMQProducer.shutdown();
        }
    }


    @Override
    public SendResult send(Message message, MessageQueueSelector selector, Object arg) {
        SendResult sendResult;
        try {
            sendResult = this.defaultMQProducer.send(message, selector, arg);
        } catch (Exception e) {
            throw new AsyncException("defaultMQProducer send order exception", e);
        }
        return sendResult;
    }


    @Override
    public boolean isStarted() {
        return this.started.get();
    }


    @Override
    public boolean isClosed() {
        return this.closed.get();
    }
}


